package com.hz.ruoyimqttconsume.handler;


import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.hz.ruoyimqttconsume.domain.AgrEnvironmentalData;
import com.hz.ruoyimqttconsume.domain.AgrSoilData;
import com.hz.ruoyimqttconsume.utils.LatestDataCache;
import com.ruoyi.common.core.utils.DateUtils;
import com.ruoyi.system.api.EquipmentService;
import com.ruoyi.system.api.domain.AgrAlertLinkResultVo;
import com.ruoyi.system.api.domain.AgrEquipmentFertigation;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author hwb zjk
 * description: MQTT消息处理
 */
@AllArgsConstructor
@Component
@Slf4j
public class MQTTMessageHandler implements MessageHandler {

    @Autowired
    private SimpMessagingTemplate messagingTemplate;

    @Autowired
    private EquipmentService equipmentService;

    @Autowired
    private LatestDataCache dataCache;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private ThreadPoolTaskExecutor taskExecutor; // 注入线程池


    @Override
    public void handleMessage(Message<?> message) throws MessagingException {
        String topic = (String) message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC);

        // 将消息处理逻辑放入线程池中执行
        taskExecutor.execute(() -> {
            try {
                processMessage(topic, message);
            } catch (Exception e) {
                log.error("Error processing MQTT message", e);
            }
        });
    }

    private void processMessage(String topic, Message<?> message) {
        List<AgrAlertLinkResultVo> list = new ArrayList<>();

        if (topic.equals("base/data/pest")) {
            log.info("收到虫情设备实时数据为--->{}:", message.getPayload());
            dataCache.updatePestData(message.getPayload());
            messagingTemplate.convertAndSend("/topic/AgrPestData", message.getPayload());
        } else if (topic.equals("base/data/soil")) {
            log.info("收到土壤实时数据为--->{}", message.getPayload());
            AgrSoilData agrSoilData = JSON.parseObject(message.getPayload().toString(), AgrSoilData.class);
            processSoilData(agrSoilData);
            dataCache.updateSoilData(message.getPayload());
            messagingTemplate.convertAndSend("/topic/AgrSoilData", message.getPayload());
        } else if (topic.equals("base/data/environmental")) {
            AgrEnvironmentalData agrEnvironmentalData = JSON.parseObject(message.getPayload().toString(), AgrEnvironmentalData.class);
            log.info("收到气象设备实时数据为--->{}:", message.getPayload());
            processEnvironmentalData(agrEnvironmentalData);
            dataCache.updateEnvData(message.getPayload());
            messagingTemplate.convertAndSend("/topic/AgrEnvironmentalData", message.getPayload());
        }
    }

    /**
     * 处理土壤数据
     */
    private void processSoilData(AgrSoilData agrSoilData) {
        List<AgrAlertLinkResultVo> list = getAlertRules(agrSoilData.getEquipmentId());
        log.info("土壤设备阈值规则数据--->{}:", list.toString());

        for (AgrAlertLinkResultVo agrAlertLinkResultVo : list) {
            int exeCondition = agrAlertLinkResultVo.getExeCondition();
            String fieldName = agrAlertLinkResultVo.getFieldName();
            double select = 0.0;

            switch (fieldName) {
                case "土壤酸碱度":
                    select = agrSoilData.getSoilPh();
                    break;
                case "土壤湿度":
                    select = agrSoilData.getSoilMoisture();
                    break;
                case "土壤温度":
                    select = agrSoilData.getSoilTemperature();
                    break;
                case "土壤盐度":
                    select = agrSoilData.getSoilSalinity();
                    break;
            }

            pd(exeCondition, agrAlertLinkResultVo, select);
        }
    }

    /**
     * 处理气象数据
     */
    private void processEnvironmentalData(AgrEnvironmentalData agrEnvironmentalData) {
        List<AgrAlertLinkResultVo> list = getAlertRules(agrEnvironmentalData.getEquipmentId());
        log.info("气象设备阈值规则数据--->{}:", list.toString());

        for (AgrAlertLinkResultVo agrAlertLinkResultVo : list) {
            int exeCondition = agrAlertLinkResultVo.getExeCondition();
            String fieldName = agrAlertLinkResultVo.getFieldName();
            double select = 0.0;

            switch (fieldName) {
                case "大气温度":
                    select = agrEnvironmentalData.getAtmoTemperature();
                    break;
                case "大气湿度":
                    select = agrEnvironmentalData.getAtmoHumidity();
                    break;
                case "二氧化碳浓度":
                    select = agrEnvironmentalData.getCooDensity();
                    break;
                case "光照强度":
                    select = agrEnvironmentalData.getEnvLight();
                    break;
                case "大气压强":
                    select = agrEnvironmentalData.getAtmoPressure();
                    break;
            }

            pd(exeCondition, agrAlertLinkResultVo, select);
        }
    }

    /**
     * 获取设备阈值规则
     */
    private List<AgrAlertLinkResultVo> getAlertRules(Long equipmentId) {
        List<AgrAlertLinkResultVo> list = new ArrayList<>();
        ObjectMapper objectMapper = new ObjectMapper();
        Object data = equipmentService.getAllBySourceId(equipmentId).get("data");

        if (data instanceof List) {
            List<Map<String, Object>> mapList = (List<Map<String, Object>>) data;
            for (Map<String, Object> map : mapList) {
                try {
                    String json = objectMapper.writeValueAsString(map);
                    AgrAlertLinkResultVo resultVo = objectMapper.readValue(json, AgrAlertLinkResultVo.class);
                    list.add(resultVo);
                } catch (Exception e) {
                    log.error("JSON 转换失败", e);
                }
            }
        }

        return list;
    }

    /**
     * 判断是否触发预警，并执行设备逻辑
     */
    public Boolean pd(int exeCondition, AgrAlertLinkResultVo agrAlertLinkResultVo, double select) {
        boolean flag = false;

        if (exeCondition == 0) {
            if (select < Double.parseDouble(agrAlertLinkResultVo.getMinNum()) || select > Double.parseDouble(agrAlertLinkResultVo.getMaxNum())) {
                flag = true;
            }
        } else if (exeCondition % 2 == 0) {
            if (select < Double.parseDouble(agrAlertLinkResultVo.getMinNum())) {
                flag = true;
            }
        } else if (exeCondition % 2 == 1) {
            if (select > Double.parseDouble(agrAlertLinkResultVo.getMaxNum())) {
                flag = true;
            }
        }

        if (flag) {

            String alertInfo = agrAlertLinkResultVo.getSourceEquipmentName() +
                    "所在地块" + agrAlertLinkResultVo.getPlotsName() +
                    agrAlertLinkResultVo.getFieldName() +
                    "异常，阈值" + agrAlertLinkResultVo.getMinNum() +
                    "-" + agrAlertLinkResultVo.getMaxNum() + agrAlertLinkResultVo.getFieldUnit() +
                    "，采集值" + select + agrAlertLinkResultVo.getFieldUnit();

            // 判断是否有执行设备
            if (agrAlertLinkResultVo.getTargetId() != null) {
                if (agrAlertLinkResultVo.getTargetType().equals("水肥一体机")) {
                    if (agrAlertLinkResultVo.getTargetStatus() > 0) {


                        alertInfo =  executeFertigation(agrAlertLinkResultVo,select,alertInfo);


                    } else {
                        flag=false;
                        alertInfo += "失败，目标设备不在线，请手动控制";
                    }
                } else if (agrAlertLinkResultVo.getTargetType().equals("环境调控")) {
                    if (agrAlertLinkResultVo.getTargetStatus() > 0) {
                     // Integer i=  equipmentService.getAgrEnvRegulationByEquipmentIdInner(agrAlertLinkResultVo.getTargetId());
                         alertInfo=executeEnvRegulation(agrAlertLinkResultVo,alertInfo);

                    } else {
                        flag=false;
                        alertInfo += "失败，目标设备不在线，请手动控制";
                    }
                }
            }
            if (flag) {
                // 发送预警信息到 RabbitMQ
                sendAlertToRabbitMQ(agrAlertLinkResultVo.getSourceId().toString(), alertInfo, agrAlertLinkResultVo);
            }

        }

        return false;
    }

    /**
     * 执行水肥一体机逻辑
     */
    private String executeFertigation(AgrAlertLinkResultVo agrAlertLinkResultVo,Double select,String alertInfo) {
        Integer sfStatus = equipmentService.getSfStatusByEquipmentIdInner(agrAlertLinkResultVo.getTargetId());
        AgrEquipmentFertigation agrEquipmentFertigation = new AgrEquipmentFertigation();
        if (sfStatus == 0 && agrAlertLinkResultVo.getTargetStatus2() > 0) {


            Double d = Double.parseDouble(agrAlertLinkResultVo.getExeTime()) / 60;
            agrEquipmentFertigation.setIrrigationTime(d.toString());
            agrEquipmentFertigation.setFertSolution(0);
            agrEquipmentFertigation.setWaterInjection((int) (Integer.parseInt(agrAlertLinkResultVo.getExeTime()) * 8.333));

            if (agrAlertLinkResultVo.getFieldName().equals("土壤酸碱度") ||
                    (agrAlertLinkResultVo.getFieldName().equals("土壤盐度") && select < Double.parseDouble(agrAlertLinkResultVo.getMinNum()))) {
                agrEquipmentFertigation.setFertSolution(10);
                agrEquipmentFertigation.setInstantFerti(100.0);
            }

            agrEquipmentFertigation.setEquipmentId(agrAlertLinkResultVo.getTargetId());
            agrEquipmentFertigation.setMainPump(1);
            List<Long> plotsId = new ArrayList<>();
            plotsId.add(agrAlertLinkResultVo.getPlotsId());
            agrEquipmentFertigation.setPlotsId(plotsId);
            equipmentService.updateAgrEquipmentFertigation(agrEquipmentFertigation);
            alertInfo += ",自动开启" + agrAlertLinkResultVo.getTargetEquipmentName();
            agrEquipmentFertigation.setCountdownHours(d);
            agrEquipmentFertigation.setFertUse(Long.valueOf(agrEquipmentFertigation.getFertSolution()));
            agrEquipmentFertigation.setWaterUse(Long.valueOf(agrEquipmentFertigation.getWaterInjection()));
            agrEquipmentFertigation.setInletPressure(100.0);
            agrEquipmentFertigation.setInstantWater(100.0);
            equipmentService.scheduleCountdown(agrEquipmentFertigation);
        }else if (sfStatus == 1 && agrAlertLinkResultVo.getTargetStatus2() == 0) {
            alertInfo += ",自动关闭" + agrAlertLinkResultVo.getTargetEquipmentName();
            agrEquipmentFertigation.setEquipmentId(agrAlertLinkResultVo.getTargetId());
            agrEquipmentFertigation.setMainPump(0);
            equipmentService.updateAgrEquipmentFertigation(agrEquipmentFertigation);
        }
        return alertInfo;
    }

    /**
     * 执行环境调控逻辑
     */
    private String executeEnvRegulation(AgrAlertLinkResultVo agrAlertLinkResultVo,String alertInfo) {

            HashMap<String, Object> params = new HashMap<>();
            params.put("equipmentId", agrAlertLinkResultVo.getTargetId());

            if (agrAlertLinkResultVo.getTargetStatus2() == 1) {
                params.put("status", 1);
                equipmentService.updateStatus(params);
                alertInfo += ",自动开启" + agrAlertLinkResultVo.getTargetEquipmentName();
                equipmentService.scheduleCountdownEq(Double.parseDouble(agrAlertLinkResultVo.getExeTime()), agrAlertLinkResultVo.getTargetId());

            }else {
                params.put("status", 0);
                alertInfo += ",自动关闭" + agrAlertLinkResultVo.getTargetEquipmentName();
                equipmentService.updateStatus(params);
            }



        return alertInfo;

    }

    /**
     * 发送预警信息到 RabbitMQ
     */
    private void sendAlertToRabbitMQ(String code, String alertInfo, AgrAlertLinkResultVo agrAlertLinkResultVo) {
        Map<String, Object> message = new HashMap<>();
        message.put("code", code);
        message.put("alertInfo", alertInfo);
//        message.put("agrAlertLinkResultVo", agrAlertLinkResultVo);

        // 及时发送信息
        rabbitTemplate.convertAndSend("alert.exchange", "alert.queue", message);
        log.info("预警信息已发送到普通队列: {}", message);

        message.remove("code");

        message.put("exeTime", DateUtils.getTime());
        message.put("agrAlertLinkResultVo", agrAlertLinkResultVo);
//        // 根据设备执行时间动态设置 TTL
        if (agrAlertLinkResultVo.getExeTime()==null) {
            rabbitTemplate.convertAndSend("delay.exchange", "delay.queue", message, msg -> {
                msg.getMessageProperties().setExpiration("100"); // 设置 TTL
                return msg;
            });
        }else {
            int delayTime = (int)Double.parseDouble(agrAlertLinkResultVo.getExeTime())*60*1000; // 设备执行时间（单位：毫秒）

            rabbitTemplate.convertAndSend("delay.exchange", "delay.queue", message, msg -> {
                msg.getMessageProperties().setExpiration(String.valueOf(delayTime)); // 设置 TTL
                return msg;

            });}
        log.info("预警已发送到延时队列: {}", message);



    }
}
